package com.study.storm.demo2;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

/**
 * @author Mayjean
 * @Date2020/8/24 21:20
 */
public class TopologyMain {

    public static void main(String[] args) throws InterruptedException {
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        //parallelism_hint 执行线程数 setNumTasks 所有线程运行任务总数，以下配置是2个spout线程各自运行一个任务
        topologyBuilder.setSpout("worldCountSpout", new WordReader(), 1).setNumTasks(1);
        //topologyBuilder.setBolt("worldCutBolt", new WorldCutRichBolt(), 1).shuffleGrouping("worldCountSpout");
        //tuple随机分发给下一阶段的bolt ; parallelism_hint 执行线程数  ;
        // setNumTasks 所有线程运行任务总数，以下配置是2个线程各自运行一个Bolt任务
        topologyBuilder.setBolt("worldCutBolt", new WordNormalizer(), 1)
                .setNumTasks(1)
                .shuffleGrouping("worldCountSpout");
        //tuple按字段char的值分发给下一阶段的bolt
//        topologyBuilder.setBolt("worldCountBolt", new WordCounter(), 2)
//                .fieldsGrouping("worldCutBolt", new Fields("word"));
        Config config = new Config();
        //关闭ack
        //conf.setNumAckers(0);
        config.put("wordsFile", "words.txt");
        //本地模式
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("myTopology", config, topologyBuilder.createTopology());
        //  关闭本地集群
        Thread.sleep(10000);
        cluster.shutdown();
    }


    public static void s(String[] args)throws InterruptedException {
        //定义拓扑
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("word-reader", new WordReader());
        builder.setBolt("word-normalizer", new WordNormalizer()).shuffleGrouping("word-reader");
        builder.setBolt("word-counter", new WordCounter(),2)
                .fieldsGrouping("word-normalizer", new Fields("word"));

        //配置
        Config config = new Config();
        config.put("wordsFile", "words.txt");
        config.setDebug(false);
        //运行拓扑
        //config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("myTopology", config, builder.createTopology());
        //  关闭本地集群
        Thread.sleep(10000);
        cluster.shutdown();
    }

}
